package ma.zhiqiang.dataline.connector.transform;

import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONUtil;
import ma.zhiqiang.dataline.core.dto.*;
import ma.zhiqiang.dataline.core.engine.Transform;
import ma.zhiqiang.dataline.util.HandlerUtil;

import java.util.List;
import java.util.Properties;

/**
 * 将一条记录转成单元格集合，每个单元格对应一个字段，包含字段名称、字段类型、字段值
 */
public class ColumnsTransform implements Transform {
    Properties properties;
    List<Schema> schemas;
    public ColumnsTransform(Properties properties) {
        this.properties = properties;
        this.schemas = JSONUtil.toList((JSONArray) properties.get("schema"), Schema.class);
    }

    @Override
    public SinkRecord transfer(SourceRecord sourceRecord) {
        ColumnMap columnMap = new ColumnMap<>(schemas.size());
        schemas.stream().forEach(schema -> {
            Column column = sourceRecord.getColumn(schema.getAlias());
            if(null!=schema.getHandler()){
                column = HandlerUtil.columnFieldHandle(schema.getHandler(), column);
            }
            columnMap.put(schema.getAlias(),column);
        });
        SinkRecord sinkRecord = new SinkRecord(ColumnMap.class,columnMap);
        return sinkRecord;
    }

}
